package com.wudl.hudi.source;

import com.alibaba.fastjson.JSON;

import com.sun.scenario.effect.impl.sw.sse.SSEBlend_SRC_OUTPeer;
import com.wudl.hudi.entity.Order;
import com.wudl.hudi.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.text.SimpleDateFormat;
import java.util.*;

/**
 * @author ：wudl
 * @date ：Created in 2022-01-01 13:05
 * @description： 产生数据
 * @modified By：
 * @version: 1.0
 */

public class GenerateData implements SourceFunction<String>  {


        private boolean isRunning = true;

        String[] citys = {"北京","广东","山东","江苏","河南","上海","河北","浙江","香港","山西","陕西","湖南","重庆","福建","天津","云南","四川","广西","安徽","海南","江西","湖北","山西","辽宁","内蒙古"};

        Integer i = 0;
        List<Order>  list = new ArrayList<>();
        @Override
        public void run(SourceContext<String> ctx) throws Exception {

            Random random = new Random();
            SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            while (isRunning) {
                int number = random.nextInt(4) + 1;
                Integer id =  i++;
                String name = df.format(new Date());
//                Integer name = id+100;
                String address = "1";
                String city = citys[random.nextInt(citys.length)];
//                int age = random.nextInt(25);

                String orderId = getDateOrder();
                String userId = number+"";
                String orderTime = getDate();
              String ip =  getRandomIp();


                double a = Math.random()* (100);
                String substring = String.valueOf(a).substring(0, 5);
              double orderMoney = Double.valueOf(substring);
              int orderStatus = 0;

                String phone = getTel();
                Order  order = new Order( orderId,  userId,  orderTime,  ip,  orderMoney,  orderStatus);
//                list.add(dorisBean);
//                if (list.size()==2)
//                {
//                    Map<String,List> map = new HashMap<>();
//                    map.put("data",list);
                    String s = JSON.toJSONString(order);
                    System.out.println("map--->"+s);
                    list = new ArrayList<>();
                    ctx.collect(s);
                    list.clear();
                    Thread.sleep(5000);
//                }


            }
        }

        @Override
        public void cancel() {

            isRunning = false;
        }


    private static String[] telFirst="134,135,136,137,138,139,150,151,152,157,158,159,130,131,132,155,156,133,153".split(",");
    private static String getTel() {
        int index=getNum(0,telFirst.length-1);
        String first=telFirst[index];
        String second=String.valueOf(getNum(1,888)+10000).substring(1);
        String third=String.valueOf(getNum(1,9100)+10000).substring(1);
        return first+second+third;
    }

    /**
     * 获取当前时间
     * @return
     */
    public static String getDate() throws InterruptedException {
        Calendar calendar = Calendar.getInstance();
        Date date = calendar.getTime();
        String dataStr = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SSS").format(date);
        Thread.sleep(10);
        return dataStr;
    }

    static  int orderaTmp = 1000;
    public static String getDateOrder() throws InterruptedException {

        Calendar calendar = Calendar.getInstance();
        Date date = calendar.getTime();
        String dataStr = new SimpleDateFormat("yyyyMMddHHmmssSSS").format(date);
        Thread.sleep(10);
        orderaTmp +=2;
        return dataStr+orderaTmp;
    }
    /*
     * 随机生成国内IP地址
     */
    public static String getRandomIp(){

        //ip范围
        int[][] range = {{607649792,608174079},//36.56.0.0-36.63.255.255
                {1038614528,1039007743},//61.232.0.0-61.237.255.255
                {1783627776,1784676351},//106.80.0.0-106.95.255.255
                {2035023872,2035154943},//121.76.0.0-121.77.255.255
                {2078801920,2079064063},//123.232.0.0-123.235.255.255
                {-1950089216,-1948778497},//139.196.0.0-139.215.255.255
                {-1425539072,-1425014785},//171.8.0.0-171.15.255.255
                {-1236271104,-1235419137},//182.80.0.0-182.92.255.255
                {-770113536,-768606209},//210.25.0.0-210.47.255.255
                {-569376768,-564133889}, //222.16.0.0-222.95.255.255
        };

        Random rdint = new Random();
        int index = rdint.nextInt(10);
        String ip = num2ip(range[index][0]+new Random().nextInt(range[index][1]-range[index][0]));
        return ip;
    }
    /*
     * 将十进制转换成ip地址
     */
    public static String num2ip(int ip) {
        int [] b=new int[4] ;
        String x = "";

        b[0] = (int)((ip >> 24) & 0xff);
        b[1] = (int)((ip >> 16) & 0xff);
        b[2] = (int)((ip >> 8) & 0xff);
        b[3] = (int)(ip & 0xff);
        x=Integer.toString(b[0])+"."+Integer.toString(b[1])+"."+Integer.toString(b[2])+"."+Integer.toString(b[3]);

        return x;
    }

    public static int getNum(int start,int end) {
        return (int)(Math.random()*(end-start+1)+start);
    }

    public static void main(String[] args) throws Exception {
         String default_topic = "hudiflink";
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(8);
        DataStreamSource<String> generateSource = env.addSource(new GenerateData());
        generateSource.print("--------");
        generateSource.addSink(MyKafkaUtil.getKafkaProducer(default_topic));

        env.execute();


    }

}
